package com.swak.reactivex.threads;

import java.util.Comparator;
import java.util.PriorityQueue;
import java.util.Queue;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.Executor;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicIntegerFieldUpdater;

import com.swak.reactivex.threads.waitstrategy.YieldingWaitStrategy;
import com.swak.utils.Queues;

import io.netty.util.internal.ObjectUtil;
import io.netty.util.internal.PlatformDependent;

/**
 * SingleThreadEventLoop.
 * 
 * @author DELL
 */
public class SingleThreadEventLoop extends AbstractEventLoop {

	static final Runnable WAKEUP_TASK = new Runnable() {
		@Override
		public void run() {
		} // Do nothing
	};

	private static final AtomicIntegerFieldUpdater<SingleThreadEventLoop> STATE_UPDATER = AtomicIntegerFieldUpdater
			.newUpdater(SingleThreadEventLoop.class, "state");
	private static final Comparator<ScheduledFutureTask<?>> SCHEDULED_FUTURE_TASK_COMPARATOR = new Comparator<ScheduledFutureTask<?>>() {
		@Override
		public int compare(ScheduledFutureTask<?> o1, ScheduledFutureTask<?> o2) {
			return o1.compareTo(o2);
		}
	};

	private static final int ST_NOT_STARTED = 1;
	private static final int ST_STARTED = 2;
	private static final int ST_SHUTTING_DOWN = 3;
	private static final int ST_SHUTDOWN = 4;
	private static final int ST_TERMINATED = 5;

	private final CountDownLatch threadLock = new CountDownLatch(1);
	private final Queue<Runnable> taskQueue;
	long nextTaskId;
	private final PriorityQueue<ScheduledFutureTask<?>> scheduledTaskQueue;
	private volatile Thread thread;
	private volatile boolean interrupted;
	private final Executor executor;
	private final WaitStrategy waitStrategy;
	private final Barrier barrier;
	private volatile int state = ST_NOT_STARTED;

	private volatile long gracefulShutdownQuietPeriod;
	private volatile long gracefulShutdownTimeout;
	private long gracefulShutdownStartTime;
	private long lastExecutionTime;
	private final CompletableFuture<Void> terminationFuture = new CompletableFuture<>();

	public SingleThreadEventLoop() {
		this(new SwakThreadFactory("EventLoop", false), new YieldingWaitStrategy());
	}

	public SingleThreadEventLoop(WaitStrategy waitStrategy) {
		this(new SwakThreadFactory("EventLoop", false), waitStrategy);
	}

	public SingleThreadEventLoop(ThreadFactory threadFactory, WaitStrategy waitStrategy) {
		ObjectUtil.checkNotNull(threadFactory, "task");
		this.executor = (run) -> {
			threadFactory.newThread(run).start();
		};
		this.scheduledTaskQueue = new PriorityQueue<>(11, SCHEDULED_FUTURE_TASK_COMPARATOR);
		this.taskQueue = Queues.newMpscQueue();
		this.waitStrategy = waitStrategy;
		this.barrier = () -> {
			return this.isShuttingDown() ? 0 : this.hasTasks();
		};
	}

	long hasTasks() {
		if (!taskQueue.isEmpty()) {
			return 0;
		}
		ScheduledFutureTask<?> task = peekScheduledTask();
		if (task != null) {
			return task.delayNanos();
		}
		return Long.MAX_VALUE;
	}

	@Override
	public void execute(Runnable task) {
		ObjectUtil.checkNotNull(task, "task");
		execute(task, !(task instanceof LazyRunnable));
	}

	@Override
	public void runOnEventLoop(Runnable command) {
		boolean inEventLoop = this.inEventLoop();
		if (inEventLoop) {
			safeExecute(command);
		} else {
			execute(command);
		}
	}

	@Override
	public void lazyExecute(Runnable task) {
		execute(ObjectUtil.checkNotNull(task, "task"), false);
	}

	@Override
	public ScheduledFutureTask<?> schedule(Runnable command, long delay, TimeUnit unit) {
		ObjectUtil.checkNotNull(command, "command");
		ObjectUtil.checkNotNull(unit, "unit");
		if (delay < 0) {
			delay = 0;
		}
		return schedule(
				new ScheduledFutureTask<Void>(this, command, ScheduledFutureTask.deadlineNanos(unit.toNanos(delay))));
	}

	@Override
	public ScheduledFuture<?> scheduleAtFixedRate(Runnable command, long initialDelay, long period, TimeUnit unit) {
		ObjectUtil.checkNotNull(command, "command");
		ObjectUtil.checkNotNull(unit, "unit");
		if (initialDelay < 0) {
			throw new IllegalArgumentException(String.format("initialDelay: %d (expected: >= 0)", initialDelay));
		}
		if (period <= 0) {
			throw new IllegalArgumentException(String.format("period: %d (expected: > 0)", period));
		}
		return schedule(new ScheduledFutureTask<Void>(this, command,
				ScheduledFutureTask.deadlineNanos(unit.toNanos(initialDelay)), unit.toNanos(period)));
	}

	private <V> ScheduledFutureTask<V> schedule(final ScheduledFutureTask<V> task) {
		if (inEventLoop()) {
			scheduleFromEventLoop(task);
		} else {
			execute(task);
		}
		return task;
	}

	final void removeScheduled(ScheduledFutureTask<?> task) {
		assert task.isCancelled();
		if (inEventLoop()) {
			this.scheduledTaskQueue.remove(task);
		} else {
			// task will remove itself from scheduled task queue when it runs
			lazyExecute(task);
		}
	}

	final void scheduleFromEventLoop(ScheduledFutureTask<?> task) {
		this.scheduledTaskQueue.add(task.setId(++nextTaskId));
	}

	private void execute(Runnable task, boolean immediate) {
		boolean inEventLoop = inEventLoop();
		addTask(task);
		if (!inEventLoop) {
			startThread();
			if (isShutdown()) {
				boolean reject = false;
				try {
					if (removeTask(task)) {
						reject = true;
					}
				} catch (UnsupportedOperationException e) {
					// The task queue does not support removal so the best thing we can do is to
					// just move on and
					// hope we will be able to pick-up the task before its completely terminated.
					// In worst case we will log on termination.
				}
				if (reject) {
					reject();
				}
			}
		}

		if (immediate) {
			wakeup(inEventLoop);
		}
	}

	protected void wakeup(boolean inEventLoop) {
		if (!inEventLoop) {
			this.waitStrategy.wakeup();
		}
	}

	protected boolean removeTask(Runnable task) {
		return taskQueue.remove(ObjectUtil.checkNotNull(task, "task"));
	}

	protected void addTask(Runnable task) {
		ObjectUtil.checkNotNull(task, "task");
		if (!offerTask(task)) {
			reject(task);
		}
	}

	final boolean offerTask(Runnable task) {
		if (isShutdown()) {
			reject();
		}
		return taskQueue.offer(task);
	}

	private void startThread() {
		if (state == ST_NOT_STARTED) {
			if (STATE_UPDATER.compareAndSet(this, ST_NOT_STARTED, ST_STARTED)) {
				boolean success = false;
				try {
					doStartThread();
					success = true;
				} finally {
					if (!success) {
						STATE_UPDATER.compareAndSet(this, ST_STARTED, ST_NOT_STARTED);
					}
				}
			}
		}
	}

	private void doStartThread() {
		assert thread == null;
		this.executor.execute(() -> {
			thread = Thread.currentThread();
			if (interrupted) {
				thread.interrupt();
			}

			boolean success = false;
			lastExecutionTime = ScheduledFutureTask.nanoTime();
			try {
				SingleThreadEventLoop.this.run();
				success = true;
			} catch (Exception t) {
				logger.warn("Unexpected exception from an event executor: ", t);
			} finally {
				for (;;) {
					int oldState = state;
					if (oldState >= ST_SHUTTING_DOWN
							|| STATE_UPDATER.compareAndSet(SingleThreadEventLoop.this, oldState, ST_SHUTTING_DOWN)) {
						break;
					}
				}

				// Check if confirmShutdown() was called at the end of the loop.
				if (success && gracefulShutdownStartTime == 0) {
					if (logger.isErrorEnabled()) {
						logger.error("Buggy " + SingleThreadEventLoop.class.getSimpleName() + ".confirmShutdown() must "
								+ "be called before run() implementation terminates.");
					}
				}

				try {
					// Run all remaining tasks and shutdown hooks. At this point the event loop
					// is in ST_SHUTTING_DOWN state still accepting tasks which is needed for
					// graceful shutdown with quietPeriod.
					for (;;) {
						if (confirmShutdown()) {
							break;
						}
					}

					// Now we want to make sure no more tasks can be added from this point. This is
					// achieved by switching the state. Any new tasks beyond this point will be
					// rejected.
					for (;;) {
						int oldState = state;
						if (oldState >= ST_SHUTDOWN
								|| STATE_UPDATER.compareAndSet(SingleThreadEventLoop.this, oldState, ST_SHUTDOWN)) {
							break;
						}
					}

					// We have the final set of tasks in the queue now, no more can be added, run
					// all remaining.
					// No need to loop here, this is the final pass.
					confirmShutdown();
				} finally {
					try {
						cleanup();
					} finally {
						STATE_UPDATER.set(SingleThreadEventLoop.this, ST_TERMINATED);
						threadLock.countDown();
						int numUserTasks = drainTasks();
						if (numUserTasks > 0 && logger.isWarnEnabled()) {
							logger.warn("An event executor terminated with " + "non-empty task queue (" + numUserTasks
									+ ')');
						}
						terminationFuture.complete(null);
					}
				}
			}

		});
	}

	final int drainTasks() {
		int numTasks = 0;
		for (;;) {
			Runnable runnable = taskQueue.poll();
			if (runnable == null) {
				break;
			}
			// WAKEUP_TASK should be just discarded as these are added internally.
			// The important bit is that we not have any user tasks left.
			if (WAKEUP_TASK != runnable) {
				numTasks++;
			}
		}
		return numTasks;
	}

	protected boolean confirmShutdown() {
		if (!isShuttingDown()) {
			return false;
		}

		if (!inEventLoop()) {
			throw new IllegalStateException("must be invoked from an event loop");
		}

		cancelScheduledTasks();

		if (gracefulShutdownStartTime == 0) {
			gracefulShutdownStartTime = ScheduledFutureTask.nanoTime();
		}

		if (runAllTasks()) {
			if (isShutdown()) {
				// Executor shut down - no new tasks anymore.
				return true;
			}

			// There were tasks in the queue. Wait a little bit more until no tasks are
			// queued for the quiet period or
			// terminate if the quiet period is 0.
			// See https://github.com/netty/netty/issues/4241
			if (gracefulShutdownQuietPeriod == 0) {
				return true;
			}
			taskQueue.offer(WAKEUP_TASK);
			return false;
		}

		final long nanoTime = ScheduledFutureTask.nanoTime();

		if (isShutdown() || nanoTime - gracefulShutdownStartTime > gracefulShutdownTimeout) {
			return true;
		}

		if (nanoTime - lastExecutionTime <= gracefulShutdownQuietPeriod) {
			// Check if any tasks were added to the queue every 100ms.
			taskQueue.offer(WAKEUP_TASK);
			try {
				Thread.sleep(100);
			} catch (InterruptedException e) {
				// Ignore
			}

			return false;
		}

		// No tasks were added for last quiet period - hopefully safe to shut down.
		// (Hopefully because we really cannot make a guarantee that there will be no
		// execute() calls by a user.)
		return true;
	}

	protected void cancelScheduledTasks() {
		assert inEventLoop();
		PriorityQueue<ScheduledFutureTask<?>> scheduledTaskQueue = this.scheduledTaskQueue;
		if (scheduledTaskQueue == null || scheduledTaskQueue.isEmpty()) {
			return;
		}
		final ScheduledFutureTask<?>[] scheduledTasks = scheduledTaskQueue.toArray(new ScheduledFutureTask<?>[0]);
		for (ScheduledFutureTask<?> task : scheduledTasks) {
			task.cancelWithoutRemove(false);
		}
		scheduledTaskQueue.clear();
	}

	protected void cleanup() {
	}

	@Override
	public CompletableFuture<Void> shutdownGracefully(long quietPeriod, long timeout, TimeUnit unit) {
		ObjectUtil.checkPositiveOrZero(quietPeriod, "quietPeriod");
		if (timeout < quietPeriod) {
			throw new IllegalArgumentException(
					"timeout: " + timeout + " (expected >= quietPeriod (" + quietPeriod + "))");
		}
		ObjectUtil.checkNotNull(unit, "unit");

		if (isShuttingDown()) {
			return terminationFuture();
		}

		boolean inEventLoop = inEventLoop();
		boolean wakeup;
		int oldState;
		for (;;) {
			if (isShuttingDown()) {
				return terminationFuture();
			}
			int newState;
			wakeup = true;
			oldState = state;
			if (inEventLoop) {
				newState = ST_SHUTTING_DOWN;
			} else {
				switch (oldState) {
				case ST_NOT_STARTED:
				case ST_STARTED:
					newState = ST_SHUTTING_DOWN;
					break;
				default:
					newState = oldState;
					wakeup = false;
				}
			}
			if (STATE_UPDATER.compareAndSet(this, oldState, newState)) {
				break;
			}
		}
		gracefulShutdownQuietPeriod = unit.toNanos(quietPeriod);
		gracefulShutdownTimeout = unit.toNanos(timeout);

		if (ensureThreadStarted(oldState)) {
			return terminationFuture;
		}

		if (wakeup) {
			taskQueue.offer(WAKEUP_TASK);
			wakeup(inEventLoop);
		}

		return terminationFuture();
	}

	private boolean ensureThreadStarted(int oldState) {
		if (oldState == ST_NOT_STARTED) {
			try {
				doStartThread();
			} catch (Throwable cause) {
				STATE_UPDATER.set(this, ST_TERMINATED);
				terminationFuture.completeExceptionally(cause);

				if (!(cause instanceof Exception)) {
					// Also rethrow as it may be an OOME for example
					PlatformDependent.throwException(cause);
				}
				return true;
			}
		}
		return false;
	}

	@Override
	public CompletableFuture<Void> terminationFuture() {
		return terminationFuture;
	}

	@Override
	public boolean isShuttingDown() {
		return state >= ST_SHUTTING_DOWN;
	}

	@Override
	public boolean isShutdown() {
		return state >= ST_SHUTDOWN;
	}

	@Override
	public boolean isTerminated() {
		return state == ST_TERMINATED;
	}

	@Override
	public boolean inEventLoop(Thread thread) {
		return thread == this.thread;
	}

	/**
	 * 实现真实的业务逻辑
	 */
	@Override
	protected void run() {
		for (;;) {
			try {
				boolean ranTasks = runAllTasks();
				if (!ranTasks) {
					try {
						this.waitStrategy.waitFor(this.barrier);
					} catch (Exception e) {
						// ignore
					}
				}
			} catch (Error e) {
				throw e;
			} catch (Throwable t) {
				handleLoopException(t);
			} finally {
				// Always handle shutdown even if the loop processing threw an exception.
				try {
					if (isShuttingDown()) {
						if (confirmShutdown()) {
							return;
						}
					}
				} catch (Error e) {
					throw e;
				} catch (Throwable t) {
					handleLoopException(t);
				}
			}
		}
	}

	protected boolean runAllTasks() {
		assert inEventLoop();
		boolean fetchedAll;
		boolean ranAtLeastOne = false;

		do {
			fetchedAll = fetchFromScheduledTaskQueue();
			if (runAllTasksFrom(taskQueue)) {
				ranAtLeastOne = true;
			}
		} while (!fetchedAll); // keep on processing until we fetched all scheduled tasks.

		if (ranAtLeastOne) {
			lastExecutionTime = ScheduledFutureTask.nanoTime();
		}
		return ranAtLeastOne;
	}

	private boolean fetchFromScheduledTaskQueue() {
		if (scheduledTaskQueue == null || scheduledTaskQueue.isEmpty()) {
			return true;
		}
		long nanoTime = ScheduledFutureTask.nanoTime();
		for (;;) {
			Runnable scheduledTask = pollScheduledTask(nanoTime);
			if (scheduledTask == null) {
				return true;
			}
			if (!taskQueue.offer(scheduledTask)) {
				// No space left in the task queue add it back to the scheduledTaskQueue so we
				// pick it up again.
				scheduledTaskQueue.add((ScheduledFutureTask<?>) scheduledTask);
				return false;
			}
		}
	}

	protected final Runnable pollScheduledTask(long nanoTime) {
		assert inEventLoop();

		ScheduledFutureTask<?> scheduledTask = peekScheduledTask();
		if (scheduledTask == null || scheduledTask.deadlineNanos() - nanoTime > 0) {
			return null;
		}
		scheduledTaskQueue.remove();
		scheduledTask.setConsumed();
		return scheduledTask;
	}

	final ScheduledFutureTask<?> peekScheduledTask() {
		Queue<ScheduledFutureTask<?>> scheduledTaskQueue = this.scheduledTaskQueue;
		return scheduledTaskQueue != null ? scheduledTaskQueue.peek() : null;
	}

	protected final boolean runAllTasksFrom(Queue<Runnable> taskQueue) {
		Runnable task = pollTaskFrom(taskQueue);
		if (task == null) {
			return false;
		}
		for (;;) {
			safeExecute(task);
			task = pollTaskFrom(taskQueue);
			if (task == null) {
				return true;
			}
		}
	}

	protected Runnable pollTask() {
		assert inEventLoop();
		return pollTaskFrom(taskQueue);
	}

	protected Runnable pollTaskFrom(Queue<Runnable> taskQueue) {
		for (;;) {
			Runnable task = taskQueue.poll();
			if (task != WAKEUP_TASK) {
				return task;
			}
		}
	}
}
